package me.xhy.java.springboot3.webflux.projectreactor.s3projectreactor;

import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

import java.time.Duration;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class S1Flux {

  public static void main(String[] args) throws Exception {
    // Flux 表示一个含有 N 个元素的流， Flux 是一个 Publisher

    // 1. 多元素的流
    Flux<Integer> just = Flux.just(1, 2, 3);
    just.subscribe(i -> System.out.println("i = " + i));

    System.out.println("=== === ===");
    // 2. 空流
    Flux<Object> empty = Flux.empty();
    empty.doOnComplete(() -> System.out.println("空流中没有元素，但是有结束信号"))
        .subscribe(i -> System.out.println("i = " + i));

    // 3. 不订阅流，任何事情都不会发生
    just.doOnComplete(() -> System.out.println("不会打印的"));

    just.doOnComplete(() -> System.out.println("有订阅，会打印"))
        .subscribe(i -> System.out.println("i = " + i));

    System.out.println("// 4. 事件感知 === === ===");
    // 4. 事件感知
    // 每秒产生一个数据。这里链式操作不能断开，每次操作都会产生新流，下面订阅的是对象就不是一个，事件和信号将不会产生联系
    // doOnXxx 要想要感知哪个流，就要写在那个流后面，因为每次操作都会产生新的流
    // 元素从Publisher发出，要走过所有流程后，下一个元素再发出，再走过所有流程
    // doOnNext：在每一个数据到达的时候；doOnEach： 在每个数据 和 信号 达到的时候
    Random random = new Random();
    AtomicInteger randomInt = new AtomicInteger();
    Flux<Long> interval = Flux.interval(Duration.ofSeconds(1))
        //Flux.range(1,20).delayElements(Duration.ofSeconds(1)) interval 不能接比自己频率慢的 delayElements
        .map(i -> {
          randomInt.set(random.nextInt(20));
          System.out.println("randomInt = " + randomInt.get());
          if (randomInt.get() < 3) {
            return i / 0;
          } else
            return i;
        })
        .doOnSubscribe(subscription -> System.out.println("doOnSubscribe"))
        .doOnRequest(value -> System.out.println("doOnRequest"))
        .doOnNext(aLong -> System.out.println("doOnNext = " + aLong))
        .doOnEach(longSignal -> System.out.println("doOnEach = " + longSignal))
        .doOnComplete(() -> System.out.println("doOnComplete"))
        .doOnCancel(() -> System.out.println("doOnCancel"))
        .doOnError(throwable -> System.out.println("doOnError: " + throwable.getMessage()));

    // 重写订阅者，其中包含很多 hook 函数
    interval.subscribe(new BaseSubscriber<Long>() {

      Subscription subscription;

      @Override
      protected void hookOnSubscribe(Subscription subscription) {
        // super.hookOnSubscribe(subscription);
        System.out.println("hook: hookOnSubscribe");
        this.subscription = subscription;
        this.subscription.request(1); // 要启动，背压
      }

      @Override
      protected void hookOnNext(Long value) {
        // super.hookOnNext(value);
        System.out.println("hook: hookOnNext " + value);

        if (value < 5) {
          this.subscription.request(1); // 持续流动
        } else {
          Random r = new Random();
          int i = r.nextInt(4);

          if (i == 0) {
            System.out.println("hook: hookOnNext 发起取消流");
            this.subscription.cancel(); // 条件取消流
          } else if (i == 1) {
            System.out.println("hook: hookOnNext 发生错误");
            throw new RuntimeException("发生错误");
          } else if (i == 2) {
            System.out.println("hook: hookOnNext 订阅者取消关系");
            cancel();
          } else {
            System.out.println("hook: hookOnNext 继续运行");
            this.subscription.request(1);
          }

        }
      }

      @Override
      protected void hookOnComplete() {
        super.hookOnComplete();
        System.out.println("hook: hookOnComplete");
      }

      @Override
      protected void hookOnError(Throwable throwable) {
        super.hookOnError(throwable);
        System.out.println("hook: hookOnError");
      }

      @Override
      protected void hookOnCancel() {
        // super.hookOnCancel();
        System.out.println("hook: hookOnCancel");
      }

      @Override
      protected void hookFinally(SignalType type) {
        super.hookFinally(type);
        System.out.println("hook: hookFinally");
      }

    });


    // 5. 预设的 subscribe 订阅者对流的取消
    System.out.println("5. 预设的 subscribe 订阅者对流的取消");
    Disposable subscribe = Flux.just(1, 2, 3).subscribe(System.out::println);
    subscribe.dispose();

    TimeUnit.SECONDS.sleep(20);

  }
}
